package com.sfh.service.model;

import org.reactivestreams.Publisher;

import com.jfinal.plugin.activerecord.Model;
import com.sfh.base.IResult;
import com.sfh.base.RxObserver;

import io.reactivex.Flowable;
import io.reactivex.FlowableTransformer;
import io.reactivex.annotations.NonNull;
import io.reactivex.disposables.Disposable;
import io.reactivex.schedulers.Schedulers;

@SuppressWarnings({ "rawtypes", "serial" })
public class BaseModel<M extends Model> extends Model<M> {

	/***
	 * [背压]异步请求操作
	 * 
	 * @param observable
	 * @param <T>
	 * @return
	 */
	public static <T> void execute(@NonNull Flowable<T> observable, @NonNull final IResult<T> result) {

		RxObserver<T> subscribe = new RxObserver<T>(result);
		Disposable disposable = observable.compose(new FlowableTransformer<T, T>() {

			@Override
			public Publisher<T> apply(Flowable<T> upstream) {

				return upstream.subscribeOn(Schedulers.io()).unsubscribeOn(Schedulers.io()).observeOn(Schedulers.io());
			}
		}).subscribe(subscribe, subscribe.onError, subscribe);
		subscribe.setDisposable(disposable);
	}

	/***
	 * [背压]异步请求操作
	 * 
	 * @param observable
	 * @param <T>
	 * @return
	 */
	public static <T> void execute(@NonNull Flowable<T> observable) {

		RxObserver<T> subscribe = new RxObserver<T>();
		Disposable disposable = observable.compose(new FlowableTransformer<T, T>() {

			@Override
			public Publisher<T> apply(Flowable<T> upstream) {

				return upstream.subscribeOn(Schedulers.io()).unsubscribeOn(Schedulers.io()).observeOn(Schedulers.io());
			}
		}).subscribe(subscribe, subscribe.onError, subscribe);
		subscribe.setDisposable(disposable);
	}
}
